package com.shujia.mr.kqzl1;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.ArrayList;

public class KqzlMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    public ArrayList<String> ids = new ArrayList<>();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //value: 是进入到map方法时的一行数据
        //通过观察我们文件数据发现，这一行可能是标题
        String line = value.toString();
        /*
            这一行数据可能是：
                1、包含<
                2、是标题 包含date
                3、正常的采集数据
         */
        if (line.contains("date")) {
            //需要将标题中的监测点编号进行存储
            String[] datas = line.split(",");
            for (int i = 3; i < datas.length; i++) {
                //1001A
                String id = datas[i].substring(0, datas[i].length() - 1);
                ids.add(id);
            }
        } else if (line.contains("PM2.5")) {
            String[] datas = line.split(",");
            for (int i = 0, j = 3; j < datas.length; i++, j++) {
                //处理脏数据
                if(datas[j] == null || "".equals(datas[j]) || " ".equals(datas[j]) || "\t".equals(datas[j])){
                    datas[j] = "0";
                }

                String k1 = datas[0]+"-"+ids.get(i);
                long v1 = Long.parseLong(datas[j]);

                context.write(new Text(k1),new LongWritable(v1));
            }
        }

    }
}
